
#include <iostream>
#include <sstream>
#include <stdexcept>

#include "bocat/common/ini.hpp"
#include "bocat/common/base64.hpp"
#include "fecab/packet.pb.h"

#include "lua_package.h"
#include "store_service.hpp"


namespace	fecab	{
namespace	server	{

///////////////////////////////////////////////////////////////////////
//

StoreService::StoreService( const std::string config_filename )
	:	_config_filename( config_filename ),
		_store_manager( config_filename.c_str() ),
		_store_p( NULL ),
		_L( NULL )
{
	GOOGLE_PROTOBUF_VERIFY_VERSION;

	_store_p = _store_manager.getStore();

	initializeLua();
}

StoreService::~StoreService()
{
	google::protobuf::ShutdownProtobufLibrary();

	if ( _L ) {
		lua_close( _L );
		_L = NULL;
	}

	_store_p = NULL;
}


void
StoreService::initializeLua( void )
{
	_L = lua_open();
	luaL_openlibs( _L );

	// custom library
	luaopen_package( _L );

	#define SECTION         "fecab::store"
	#define OPTION  		"script"

	bocat::common::INIFile  ini( _config_filename );

	std::string	script_filename = ini[ SECTION ][ OPTION ];
	if (script_filename.size() < 1) {
		return;
	}

	int	rc = luaL_dofile(_L, script_filename.c_str());
	if (0 != rc) {
		std::ostringstream	msg;
		msg << "fail to load LUA script " << script_filename;

		throw std::runtime_error( msg.str() );
	}
}


///////////////////////////////////////////////////////////////////////
//

int
StoreService::handle( const std::string& request_encoded, std::string& response_encoded )
{
	Packet	pkt_request;
	Packet	pkt_response;

	bool	with_response = false;

	std::string	request;
	try {
		request = bocat::common::base64_decode< std::string >( request_encoded );
	}
	catch( std::exception& e ) {
		std::ostringstream	msg;
		msg << "fail to decode base64 '" << request_encoded << "' -- " << e.what();
		throw std::runtime_error( msg.str() );
	}

	if ( !pkt_request.ParseFromString( request ) ) {
		throw std::runtime_error("Fail to parse request protocol buffer.");
	}

	switch ( pkt_request.op() ) {
	
	case	Packet::PUT :
		handlePUT( pkt_request, pkt_response );
		with_response = true;
		break;

	case	Packet::GET :
		handleGET( pkt_request, pkt_response );
		with_response = true;
		break;

	case	Packet::DELETE :
		handleDELETE( pkt_request, pkt_response );
		with_response = true;
		break;

	case	Packet::SPUT :
		handleSPUT( pkt_request, pkt_response );
		with_response = true;
		break;

	case	Packet::SMPUT :
		handleSMPUT( pkt_request, pkt_response );
		with_response = true;
		break;

	case	Packet::PING	:
		pkt_response = pkt_request;

		pkt_response.set_type( Packet::RESPONSE );
		pkt_response.set_reqid( pkt_request.sid() );
		pkt_response.set_sid( 1 );

		with_response = true;

		break;

	default	:
		std::ostringstream	msg;
		msg << "unsupport service OP : " <<  pkt_request.op();
		throw std::runtime_error( msg.str() );
		break;
	}

	if ( with_response ) {
		std::string	response;
		if ( !pkt_response.SerializeToString( &response ) ) {
			std::ostringstream	msg;
			msg << "fail to serialize response to string.";
			throw std::runtime_error( msg.str() );
		}

		response_encoded = bocat::common::base64_encode( response );
		return 1;
	}

	return	0;
}


///////////////////////////////////////////////////////////////////////
//

static  u_int32_t
getPIN( const char* symbol, u_int32_t symbol_size )
{
    u_int32_t   pin = 0;
	if (symbol_size < 1) {
		return pin;
	}

	if ( symbol_size >= (2 + sizeof(u_int32_t)) ) {
		memcpy( (char *)&pin, &symbol[2], sizeof(u_int32_t) );
	}
	else if ( symbol_size == sizeof(u_int32_t) ) {
		memcpy( (char *)&pin, symbol, sizeof(u_int32_t) );
	}
	else {
		char*	p = (char *)&pin;
		for (u_int32_t	i=0, k=0; i < sizeof(u_int32_t); ++i ) {
			p[i] = symbol[ k ];

			++k;
			if (k >= symbol_size) {
				k = 0;
			}
		}
		
	}

    return pin;
}


int
StoreService::handlePUT( Packet& pkt_request, Packet& pkt_response )
{
	pkt_response.set_type( Packet::RESPONSE );
	pkt_response.set_reqid( pkt_request.sid() );
	pkt_response.set_sid( 1 );
	pkt_response.set_op( pkt_request.op() );

	for( u_int32_t i=0; i < pkt_request.item_size(); i++ ) {
		Packet::Item*	item_p = pkt_response.add_item();

		const Packet::Item& item = pkt_request.item( i );
		std::string	 	table = item.table();
		std::string	key = item.key();
		std::string	data = item.value();

		int			rc,status = 200;
		try {
			if (table.size() < 1) {
				std::string	kpin = item.kpin();

				u_int32_t	pin = getPIN(kpin.c_str(), kpin.size());
				rc = _store_p->put( pin, key.c_str(), key.size(), data.c_str(), data.size() );
				if ( rc != 0 ) {
					status = 400;
				}

				item_p->set_kpin( kpin );
			}
			else {
				rc = _store_p->put( table.c_str(), key.c_str(), key.size(), data.c_str(), data.size() );
				if ( rc != 0 ) {
					status = 400;
				}

				item_p->set_table( table );
			}

			item_p->set_key( key );
			item_p->set_status( status );
		}
		catch(std::exception& e) {
			std::ostringstream	msg;
			msg << "fail to PUT(table:" << table << " key:" << key << ") -- " << e.what();

			status = 500;

			item_p->set_table( table );
			item_p->set_key( key );
			item_p->set_status( status );
			item_p->set_value( msg.str() );
		}
	}

	return	0;
}	//// Store::Service::handlePUT()



int
StoreService::handleGET( Packet& pkt_request, Packet& pkt_response )
{
	pkt_response.set_type( Packet::RESPONSE );
	pkt_response.set_reqid( pkt_request.sid() );
	pkt_response.set_sid( 1 );
	pkt_response.set_op( pkt_request.op() );

	for( u_int32_t i=0; i < pkt_request.item_size(); i++ ) {
		Packet::Item*	item_p = pkt_response.add_item();

		const Packet::Item& item = pkt_request.item( i );
		std::string	 	table = item.table();
		std::string	key = item.key();
		std::string	data;

		char	buf[ 512 ];
		int		rc, status = 200;
		try {
			if (table.size() < 1) {
				std::string	kpin = item.kpin();

				u_int32_t	pin = getPIN(kpin.c_str(), kpin.size());
				rc = _store_p->get( pin, key.c_str(), key.size(), buf, sizeof(buf) );
				if ( rc == 0 ) {
					status = 100;
				}
				else if ( rc < 0 ) {
					status = 400;
				}

				item_p->set_kpin( kpin );
			}
			else {
				rc = _store_p->get( table.c_str(), key.c_str(), key.size(), buf, sizeof(buf) );
				if ( rc == 0 ) {
					status = 100;
				}
				else if ( rc < 0 ) {
					status = 400;
				}

				item_p->set_table( table );
			}

			item_p->set_key( key );
			if ( rc > 0 ) {
				int	data_size = rc;
				if ( data_size > sizeof(buf) ) {
					// buffer is small
					status = 300;
					std::ostringstream	msg;
					msg << "buffer is small(buf-size:" << sizeof(buf) << " data-size:" << data_size << ")";
					item_p->set_value( msg.str() );
				}
				else {
					item_p->set_value( buf, data_size );
				}
			}
			item_p->set_status( status );
		}
		catch(std::exception& e) {
			std::ostringstream	msg;
			msg << "fail to PUT(table:" << table << " key:" << key << ") -- " << e.what();

			status = 500;

			item_p->set_table( table );
			item_p->set_key( key );
			item_p->set_status( status );
			item_p->set_value( msg.str() );
		}
	}

	return	0;
}	//// Store::Service::handleGET()



int
StoreService::handleDELETE( Packet& pkt_request, Packet& pkt_response )
{
	pkt_response.set_type( Packet::RESPONSE );
	pkt_response.set_reqid( pkt_request.sid() );
	pkt_response.set_sid( 1 );
	pkt_response.set_op( pkt_request.op() );

	for( u_int32_t i=0; i < pkt_request.item_size(); i++ ) {
		Packet::Item*	item_p = pkt_response.add_item();

		const Packet::Item& item = pkt_request.item( i );
		std::string	 	table = item.table();
		std::string	key = item.key();

		int			rc,status = 200;
		try {
			if (table.size() < 1) {
				std::string	kpin = item.kpin();

				u_int32_t	pin = getPIN(kpin.c_str(), kpin.size());
				rc = _store_p->erase( pin, key.c_str(), key.size() );
				if ( rc != 0 ) {
					status = 400;
				}

				item_p->set_kpin( kpin );
			}
			else {
				rc = _store_p->erase( table.c_str(), key.c_str(), key.size() );
				if ( rc != 0 ) {
					status = 400;
				}

				item_p->set_table( table );
			}

			item_p->set_key( key );
			item_p->set_status( status );
		}
		catch(std::exception& e) {
			std::ostringstream	msg;
			msg << "fail to PUT(table:" << table << " key:" << key << ") -- " << e.what();

			status = 500;

			item_p->set_table( table );
			item_p->set_key( key );
			item_p->set_status( status );
			item_p->set_value( msg.str() );
		}
	}

	return	0;
}	//// Store::Service::handleDELETE()


///////////////////////////////////////////////////////////////////////
//


int
StoreService::handleSPUT( Packet& pkt_request, Packet& pkt_response )
{
	pkt_response.set_type( Packet::RESPONSE );
	pkt_response.set_reqid( pkt_request.sid() );
	pkt_response.set_sid( 1 );
	pkt_response.set_op( pkt_request.op() );

	const std::string&	script = pkt_request.script();
	const std::string&	crumb  = pkt_request.crumb();
	if (script.size() < 1) {
		// no script name available
		return	0;
	}

	for( int i=0; i < pkt_request.item_size(); i++ ) {
		Packet::Item*	item_p = pkt_response.add_item();

		const Packet::Item& item = pkt_request.item( i );
		std::string	 	table = item.table();
		std::string	key = item.key();
		std::string	data = item.value();

		int			rc,status = 200;
		std::string	old_value, new_value;

		try {
			if (table.size() < 1) {
				std::string	kpin = item.kpin();

				u_int32_t	pin = getPIN(kpin.c_str(), kpin.size());
				rc = _store_p->get( pin, key, old_value );
				
				// new_value = lua_FUNCTION( table, key, old_value, data, crumb )
				rc = callLUA( table, key, old_value, data, script, crumb, new_value );
				if ( rc != 0 ) {
					status = 405;
				}
				else {
					rc = _store_p->put( pin, key, new_value );
					if ( rc != 0 ) {
						status = 400;
					}
				}

				item_p->set_kpin( kpin );
			}
			else {
				rc = _store_p->get( table, key, old_value );

				// new_value = lua_FUNCTION( table, key, old_value, data, crumb )
				rc = callLUA( table, key, old_value, data, script, crumb, new_value );
				if ( rc != 0 ) {
					status = 405;
				}
				else {
					rc = _store_p->put( table.c_str(), key, new_value );
					if ( rc != 0 ) {
						status = 400;
					}
				}

				item_p->set_table( table );
			}

			item_p->set_key( key );
			item_p->set_status( status );
			if (status == 200) {
				item_p->set_value( new_value );
			}
		}
		catch(std::exception& e) {
			std::ostringstream	msg;
			msg << "fail to PUT(table:" << table << " key:" << key << ") -- " << e.what();

			status = 500;

			item_p->set_table( table );
			item_p->set_key( key );
			item_p->set_status( status );
			item_p->set_value( msg.str() );
		}
	}

	return	0;
}	//// Store::Service::handleSPUT()


int
StoreService::handleSMPUT( Packet& pkt_request, Packet& pkt_response )
{
	return	handleSPUT( pkt_request, pkt_response );
}	//// Store::Service::handleSMPUT()


///////////////////////////////////////////////////////////////////////
//

int     
StoreService::callLUA( const std::string& table, 
					   const std::string& key, 
					   const std::string& old_value, 
					   const std::string& data, 
					   const std::string& script, 
					   const std::string& crumb, 
						     std::string& new_value )
{
	boost::mutex::scoped_lock   _scoped_lock( _mutex_lua );

	lua_getglobal(_L, script.c_str() );
	////lua_pushlstring(_L, table.c_str(), table.size() );
	//lua_pushlstring(_L, key.c_str(), key.size() );
	//lua_pushlstring(_L, old_value.c_str(), old_value.size() );
	//lua_pushlstring(_L, data.c_str(), data.size() );
	//lua_pushlstring(_L, crumb.c_str(), crumb.size() );
	lua_pushlightuserdata(_L, (void *)key.c_str() );
	lua_pushinteger(_L, (int)key.size() );
	lua_pushlightuserdata(_L, (void *)old_value.c_str() );
	lua_pushinteger(_L, (int)old_value.size() );
	lua_pushlightuserdata(_L, (void *)data.c_str() );
	lua_pushinteger(_L, (int)data.size() );
	lua_pushlightuserdata(_L, (void *)crumb.c_str() );
	lua_pushinteger(_L, (int)crumb.size() );
	
	int	rc = lua_pcall(_L, 8, 1, 0);
	if (0 != rc) {
		return	-1;
	}

	size_t	s_len = 0;
	const char*	s = lua_tolstring(_L, -1, &s_len);

	new_value.assign( s, s_len );

	lua_pop(_L, 1);

	return	0;
}



}	//// namespace	server
}	//// namespace	fecab


